package com.cl.spark.node;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.cl.spark.base.BaseSparkNode;
import com.cl.spark.dto.SparkParam;
import com.cl.spark.dto.SparkResult;
import com.cl.spark.enums.SparkNodeEnum;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.functions;
import org.springframework.stereotype.Component;
import scala.annotation.meta.param;

import java.util.List;
import java.util.Objects;

@Component
public class ContainsMatchNode extends BaseSparkNode {

    @Override
    public SparkResult process(SparkParam sparkParam) {
        JSONArray res = new JSONArray();

        List<Dataset<Row>> datasetList = sparkParam.getDatasetList();
        JSONObject expression = sparkParam.getNodeExpression();

        Dataset<Row> data1DF = datasetList.get(0);
        Dataset<Row> data2DF = datasetList.get(1);

        JSONArray fields1 = expression.getJSONArray("fields1");
        JSONArray fields2 = expression.getJSONArray("fields2");
        Dataset<Row> markedDF = data1DF.withColumn("MATCH", functions.expr(
                "exists (" +
                        "select 1 from df2 where contains(" + fields1.getString(0) + ", df2." + fields2.getString(0) + ")" +
                        ")"
        ));
        return SparkResult.success(markedDF);
    }



    @Override
    public SparkNodeEnum getType() {
        return SparkNodeEnum.MATCH;
    }
}
